package com.example.java.kafka.kafkaclients;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

public class MessageProducer {

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "43.139.50.31:9092"); // bootstrap.servers
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "PLAINTEXT"); // security.protocol
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // key.serializer
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); // value.serializer
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, String.valueOf(30 * 1000)); // max.block.ms
        props.put(ProducerConfig.ACKS_CONFIG, "all"); // acks
        // 创建 Kafka 生产者实例对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
        System.out.println("Kafka message channel created success...");
        // 循环发送消息
        for (int i = 0; i < 10; i++) {
            try {
                String sendMessage = "producer-api, id: " + i;
                ProducerRecord<String, String> producerRecord = new ProducerRecord<>("producer-api-topic",
                        "key" + (i < 5 ? 0 : 1), sendMessage);
                Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
                RecordMetadata recordMetadata = send.get();
                int partition = recordMetadata.partition();
                long offset = recordMetadata.offset();
                System.out.println("-> Message send success, content: " + sendMessage + ", partition: " + partition + ", offset: " + offset);
                Thread.sleep(3 * 1000);
            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
        }
        // 关闭生产者实例对象
        kafkaProducer.close();
    }

}

